package com.its.common.disruptor.config;

import com.its.common.disruptor.consumer.BaseMessageConsumer;
import com.its.common.disruptor.factory.RingBufferWorkerPoolFactory;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;

import java.lang.reflect.Constructor;
import java.util.List;

/**
 * @Author xiaxp
 * @Date 5/5/22 13:51
 * @Description
 */
@Getter
@Setter
@Slf4j
@EnableConfigurationProperties({DisruptorMultiConsumerConfig.class})
@ConditionalOnProperty(prefix = "its.disruptor", name = "enabled", havingValue = "true")
@Order(Ordered.LOWEST_PRECEDENCE - 1)
public class DisruptorConfigure implements CommandLineRunner, ApplicationContextAware {

    private ApplicationContext applicationContext;

    @Autowired
    private DisruptorMultiConsumerConfig disruptorMultiConsumerConfig;

    @Override
    public void run(String... args) throws Exception {
        if(null == disruptorMultiConsumerConfig){
            log.error("未定义Disruptor消费者配置信息1");
            return;
        }
        List<DisruptorConsumerProperties> confs = disruptorMultiConsumerConfig.getConfs();
        if(null != confs && confs.size() > 0){
            log.info("Disruptor准备启动，一共配置了{}类消费者", confs.size());
        }else{
            log.error("未定义Disruptor消费者配置信息2");
            return;
        }
        for (DisruptorConsumerProperties conf : confs) {
            if(null != conf){
                try {
                    Class claz = conf.getWorkerClaz();
                    if(!checkWorkerValid(claz)){
                        log.error("Disruptor消费者类必须继承BaseMessageConsumer");
                        continue;
                    }
                    Integer consumerCnt = conf.getWorkerNum();
                    if(null == consumerCnt || consumerCnt < 0){
                        continue;
                    }
                    BaseMessageConsumer[] consumers = new BaseMessageConsumer[consumerCnt];
                    for (int i = 0; i < consumers.length; i++){
                        Constructor constructor = claz.getDeclaredConstructor(String.class);
                        constructor.setAccessible(true);
                        BaseMessageConsumer consumer = (BaseMessageConsumer) constructor.newInstance("code:consumerId:"+i);
                        consumers[i] = consumer;
                    }
                    try {
                        RingBufferWorkerPoolFactory.getInstance().initAndStart(consumers);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }catch (Exception e){
                    log.error("Disruptor 初始化异常: " + e.getMessage());
                }
            }
        }
    }

    private boolean checkWorkerValid(Class worker) {
        return BaseMessageConsumer.class.isAssignableFrom(worker);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
